package com.xxxx.sink_operator;

import org.apache.flink.api.common.io.OutputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * @program: flink19Test
 * @description:
 * @author: CoreDao
 * @create: 2021-04-01 17:46
 **/

public class SinkWriteUsingOutputFormat {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        List<String> list = Arrays.asList("hello nihao", "hello nihao1", "hello nihao2");

        DataStreamSource<String> dataStreamSource = env.fromCollection(list);

        dataStreamSource.writeUsingOutputFormat(new OutputFormat<String>() {
            private String prefix = null;
            @Override
            public void configure(Configuration parameters) {
                //没有配置的话无需额外编写
            }

            /**
             * numTasks表示最大并行度
             * taskNumber为[0，numTasks]
             * @param taskNumber
             * @param numTasks
             * @throws IOException
             */
            @Override
            public void open(int taskNumber, int numTasks) throws IOException {
                System.out.println("taskNumber" + taskNumber + ", numTasks" + numTasks);
                prefix = taskNumber + 1 + ">";

            }

            @Override
            public void writeRecord(String record) throws IOException {
                System.out.println(prefix + record);
            }

            @Override
            public void close() throws IOException {
                prefix = null;
            }
        });

        System.out.println(env.getExecutionPlan());
        env.execute();
    }
}
